package cn.basttg.demo.zookeeper;

import java.io.UnsupportedEncodingException;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class ZkWebmcc implements Watcher {
	private static final String DATA_NODE = "operation";
	private ZooKeeper zk = null;
	private Integer mutex;
	private String root;

	synchronized public void process(WatchedEvent event) {
		synchronized (mutex) {
			// System.out.println("Process: " + event.getType());
			mutex.notify();
		}
	}

	/**
	 * Constructor of producer-consumer queue
	 * @param address
	 * @param name
	 */
	ZkWebmcc(String address, String name) {
		if (zk == null) {
			try {
				System.out.println("Starting ProduceService ZK:");
				zk = new ZooKeeper(address, 3000, this);
				mutex = new Integer(-1);
				System.out.println("Finished starting ProduceService ZK: " + zk);
			} catch (Exception e) {
				System.out.println(e.toString());
				zk = null;
			}
		}
		// else mutex = new Integer(-1);

		this.root = name;
		// Create ZK node name
		if (zk != null) {
			try {
				Stat s = zk.exists(root, false);
				if (s == null) {
					zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
				}
			} catch (KeeperException e) {
				System.out.println("Keeper exception when instantiating queue: " + e.toString());
			} catch (InterruptedException e) {
				System.out.println("Interrupted exception");
			}
		}
	}

	/**
	 * Add element to the queue.
	 * @param data
	 * @return
	 */
	String produce(String data) throws KeeperException, InterruptedException {
		String path = root + "/" + DATA_NODE;
		try {
			path = zk.create(path, data.getBytes("UTF-8"), Ids.OPEN_ACL_UNSAFE,
					CreateMode.PERSISTENT_SEQUENTIAL);
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		}

		return path;
	}

	public static void main(String args[]) {
		String connectionString = "172.16.160.196:2181";
		String rootPath = "/app1";

		int max = 10;
		ZkWebmcc webmcc = new ZkWebmcc(connectionString, rootPath);
		for (int i = 0; i < max; i++) {
			try {
				String path = webmcc.produce("data" + 10 + i);
				System.out.println("Produce Item" + i + ": " + path);
			} catch (KeeperException e) {
				System.out.println("Produce error: " + i);
			} catch (InterruptedException e){

            }
		}
	}
}

